package com.orderpay_detect

import java.util

import com.orderpay_detect.bean.{OrderEvent, OrderResult}
import org.apache.flink.cep.{PatternSelectFunction, PatternTimeoutFunction}
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 20:37
  *                 CEP
  *                 分类支付成功事件以及15分钟订单超时事件
  */
object OrderTimeoutWithCep {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val filePath: String = "OrderPayDetect/src/main/resources/OrderLog.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val socketStream: DataStream[String] = env.socketTextStream("192.168.88.106",7777)

    val dataStream: DataStream[OrderEvent] = socketStream.map(data => {
      val arr: Array[String] = data.split(",")
      OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
    }).assignAscendingTimestamps(_.timestamp * 1000L)

    val orderEventStream: KeyedStream[OrderEvent, Long] = dataStream.keyBy(_.orderId)

    //定义一个匹配
    val orderPayPattern: Pattern[OrderEvent, OrderEvent] = Pattern
      .begin[OrderEvent]("create").where(_.eventType == "create")
      .followedBy("pay").where(_.eventType == "pay")
      .within(Time.minutes(15))

    //将Pattern应用到数据流上，进行模式检测
    val patternStream: PatternStream[OrderEvent] = CEP.pattern(orderEventStream, orderPayPattern)


    //定义侧输出流标签，用于处理超时事件
    val orderTimeoutOutputTag = new OutputTag[OrderResult]("orderTimeout")

    //调用select方法，提取并匹配的成功支付事件以及超时事件
    val resultStream: DataStream[OrderResult] = patternStream.select(orderTimeoutOutputTag,
      new OrderTimeoutSelect,
      new OrderPaySelect)


    resultStream.print("payed")
    resultStream.getSideOutput(orderTimeoutOutputTag).print("timeout")

    env.execute("order timeout job")

  }

  class OrderTimeoutSelect extends PatternTimeoutFunction[OrderEvent, OrderResult] {
    override def timeout(map: util.Map[String, util.List[OrderEvent]], l: Long): OrderResult = {
      val timeoutOrderId = map.get("create").iterator().next().orderId
      OrderResult(timeoutOrderId, "timeout : " + l)
    }
  }

  class OrderPaySelect extends PatternSelectFunction[OrderEvent, OrderResult] {
    override def select(map: util.Map[String, util.List[OrderEvent]]): OrderResult = {
      val payedOrderId = map.get("pay").iterator().next().orderId
      OrderResult(payedOrderId, "payed successfully")
    }
  }

}
